/******************************************************************************
 * Copyright 2009 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 ******************************************************************************/

/**  Ruby extension to access sharded MySQL DBs. Provides high-level methods to
 *   perform multi-shard queries, that are executed on all shards concurrently.
 *   Designed primarily for short-lived, multiple query connections.
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>

#include <pthread.h>
#include <sys/time.h>
#include <ruby.h>

#ifdef HAVE_MYSQL_H
#include <mysql.h>
#else
#include <mysql/mysql.h>
#endif

/* Let's limit this to 1,000,000 shards... */
#define SHARD_DIGITS 6

/* Query types */
#define SINGLE_SHARD -1
#define ALL_SHARDS -2
#define UNKNOWN_SHARD -3
#define INVALID_SHARD -4

/* Error messages */
#define ERROR_QUERY_TYPE "Invalid query type specified"
#define ERROR_LIB_INIT "Error initializing MySQL library"
#define ERROR_SHARD_NUMBER "Invalid shard number"
#define ERROR_DB_CONNECT "Error connecting to database: %s"
#define ERROR_DB_QUERY "Error querying database: %s"
#define ERROR_DB_FETCH "Error retrieving results: %s"
#define ERROR_NOT_CONNECTED "Error while querying: connection not established"
#define ERROR_TOO_MANY_SHARDS "Database has too many shards"

/** @privatesection */

struct shardeddb;

/** Internal structure to hold data for each shard/thread */
struct shard_data {
  int id;
  int empty;
  int haveErrors;
  char *errorCode;
  char *errorMessage;
  MYSQL *connection;
  MYSQL_RES *results;
  struct shardeddb *db_info;
};
typedef struct shard_data shard_data_t;

/** Internal structure to hold Ruby class values */
struct shardeddb {
  int num_shards;
  char *user;
  char *password;
  char *query;
  char *host;
  char *db;
  shard_data_t *shards;
  int active_shard;
  int query_type;
};
typedef struct shardeddb shardeddb_t;

/* Global variables for the ShardedDb sub-classes */
VALUE sharded_db_error;
VALUE sharded_db_query_type;

/*******************************************************************************
 Base C work code
*******************************************************************************/

/** @privatesection */

/** Initialize MySQL library */
static void init_libmysql() {
  if (mysql_library_init(0, NULL, NULL)) {
    rb_raise(sharded_db_error, ERROR_LIB_INIT);
  }
}

/** Auxiliary function to fill in host and database names
 *  @param[in] num the shard number
 *  @param[in] host_temp hostname "template" for the several shards. Must be a
 *             string with an optional placeholder for the shard number, e.g.
 *             "dbhost%d" for dbhost0, dbhost1, etc.
 *  @param[in] db_temp database name "template" for the several shards. Must be
 *             a string with an optional placeholder for the shard number, e.g.
 *             "db%d" for db0, db1, etc.
 *  @param[out] host_ret buffer to hold the formatted host name.
 *  @param[out] db_ret buffer to hold the formatted db name
 */
static void format_names(int num, const char *host_temp, const char *db_temp,
    char *host_ret, char *db_ret) {
  if (strstr(host_temp, "%d") != NULL) {
    sprintf(host_ret, host_temp, num);
  } else {
    strcpy(host_ret, host_temp);
  }

  if (strstr(db_temp, "%d") != NULL) {
    sprintf(db_ret, db_temp, num);
  } else {
    strcpy(db_ret, db_temp);
  }
}

/** Work function for each thread: connects to its own shard. Leaves the
 *  connection open.
 *  @param[in] shard_arg shard_data_t object for the thread's shard
 *  @return NULL, always.
 */
static void *connect_job(void *shard_arg) {
  MYSQL *db = malloc(sizeof(MYSQL));
  shard_data_t *data = (shard_data_t *)shard_arg;
  char *host_name = malloc((strlen(data->db_info->host) + SHARD_DIGITS) *
      sizeof(char));
  char *db_name = malloc((strlen(data->db_info->db) + SHARD_DIGITS) *
      sizeof(char));

  /* Write host name and db name */
  format_names(data->id, data->db_info->host, data->db_info->db, host_name,
      db_name);

  mysql_init(db);

  /* Only open a connection if there isn't one already */
  if (data->connection == NULL) {
    if (!mysql_real_connect(db, host_name, data->db_info->user,
        data->db_info->password, db_name, 0, NULL, 0)) {
      data->connection = NULL;
      data->haveErrors = 1;
      data->errorCode = ERROR_DB_CONNECT;
      data->errorMessage = mysql_error(db);
    } else {
      data->haveErrors = 0;
      data->connection = db;
    }
  }

  mysql_thread_end();
  free(host_name);
  free(db_name);

  return NULL;
}

/** Work function for each thread: queries the DB, sets results, and sets
 * "empty" to false if any results are returned. Leaves the connection open.
 *  @param[in] shard_arg shard_data_t object for the thread's shard
 *  @return NULL, always.
 */
static void *query_job(void *shard_arg) {
  MYSQL_RES *results;
  unsigned int num_fields;
  shard_data_t *data = (shard_data_t *)shard_arg;
  MYSQL *db = data->connection;

  mysql_thread_init();

  if (db == NULL) {
    data->haveErrors = 1;
    data->errorCode = ERROR_NOT_CONNECTED;
    data->errorMessage = "Database is not connected";
  } else {
    if (mysql_query(db, data->db_info->query)) {
      data->haveErrors = 1;
      data->errorCode = ERROR_DB_QUERY;
      data->errorMessage = mysql_error(db);
    } else {
      results = mysql_store_result(db);
      if (!results) {
        data->haveErrors = 1;
        data->errorCode = ERROR_DB_FETCH;
        data->errorMessage = mysql_error(db);
      } else {
        data->haveErrors = 0;
        if(mysql_num_rows(results) > 0) {
          data->empty = 0;
          data->results = results;
        } else {
          data->empty = 1;
          data->results = NULL;
        }
      }
    }
  }

  mysql_thread_end();

  return NULL;
}

/** Work function for each thread: connects to its own shard, queries the DB,
 *  sets results, and sets "empty" to false if any results are returned.
 *  Leaves the connection open.
 *  @param[in] shard_arg shard_data_t object for the thread's shard
 *  @return NULL, always.
 */
static void *connect_and_query_job(void *shard_arg) {
  shard_data_t *data = (shard_data_t *)shard_arg;
  connect_job(shard_arg);
  if(!data->haveErrors) {
    query_job(shard_arg);
  }

  return NULL;
}

/** Performs a query on a single shard.
 *  @param[in] query the query to be executed on the database
 *  @param[in,out] shard shard data for the shard where the query will be
 *  executed
 */
static void query_shard(const char *query, shard_data_t *shard) {
  MYSQL_RES *results;
  unsigned int num_fields;

  if (shard->connection == NULL) {
    shard->haveErrors = 1;
    shard->errorCode = ERROR_NOT_CONNECTED;
    shard->errorMessage = "Database is not connected";
  } else {
    if (mysql_query(shard->connection, query)) {
      shard->haveErrors = 1;
      shard->errorCode = ERROR_DB_QUERY;
      shard->errorMessage = mysql_error(shard->connection);
    } else {
      results = mysql_store_result(shard->connection);
      if (!results) {
        shard->haveErrors = 1;
        shard->errorCode = ERROR_DB_FETCH;
        shard->errorMessage = mysql_error(shard->connection);
      } else {
        shard->haveErrors = 0;
        num_fields = mysql_num_fields(results);
        if(mysql_num_rows(results) > 0) {
          shard->empty = 0;
          shard->results = results;
        } else {
          shard->empty = 1;
          shard->results = NULL;
        }
      }
    }
  }
}

/** Connects to and queries all shards. Both actions are done as a single step
 *  in order to optimize overall speed.
 *  @param[in] query the query to be executed on the database
 *  @param[in,out] obj shardeddb_t struct with the object information
 *  @return first shard with results
 */
static int connect_and_query_all_shards(char *query, shardeddb_t *obj) {
  int t;
  pthread_attr_t attr;
  void *status;
  int result_shard = -1;
  pthread_t *threads = malloc(sizeof(pthread_t) * obj->num_shards);

  /* Threads should be joinable */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  obj->query = query;

  /* Start a thread for every shard */
  for(t = 0; t < obj->num_shards; t++) {
    /* Create thread to run work method */
    pthread_create(&threads[t], &attr, connect_and_query_job,
        (void *) &(obj->shards[t]));
  }

  /* Wait for all threads to finish */
  for(t = 0; t < obj->num_shards; t++) {
    pthread_join(threads[t], &status);
  }

  /* Find out which shard (if any) returned the data */
  for (t = 0; t < obj->num_shards; t++) {
    if (!obj->shards[t].empty) {
      result_shard = t;
      break;
    }
  }

  free(threads);
  return result_shard;
}

/** Connects to a single shard.
 *  @param[in,out] shard shard data for the intended shard
 */
static void connect_shard(shard_data_t *shard) {
  shard->empty = 1;
  shard->connection = NULL;
  shard->results = NULL;

  connect_job((void *) shard);
}

/** Connects to all shards.
 *  @param[in,out] obj shardeddb_t struct with the object information
 */
static void connect_all_shards(shardeddb_t *obj) {
  int t;
  pthread_attr_t attr;
  void *status;
  int result_shard = -1;
  pthread_t *threads = malloc(sizeof(pthread_t) * obj->num_shards);

  /* Threads should be joinable */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  /* Start a thread for every shard */
  for(t = 0; t < obj->num_shards; t++) {
    /* Create thread to run work method */
    pthread_create(&threads[t], &attr, connect_job, (void *) &(obj->shards[t]));
  }

  /* Wait for all threads to finish */
  for(t = 0; t < obj->num_shards; t++) {
    pthread_join(threads[t], &status);
  }

  free(threads);
}

/** Performs a query on all shards.
 *  @param[in] query the query to be executed on the database
 *  @param[in,out] obj shardeddb_t struct with the object information
 *  @return first shard with results
 */
static int query_all_shards(char *query, shardeddb_t *obj) {
  int t;
  pthread_attr_t attr;
  void *status;
  int result_shard = -1;
  pthread_t *threads = malloc(sizeof(pthread_t) * obj->num_shards);

  /* Threads should be joinable */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  obj->query = query;

  /* Start a thread for every shard */
  for(t = 0; t < obj->num_shards; t++) {
    obj->shards[t].empty = 1;
    obj->shards[t].results = NULL;
    /* Create thread to run work method */
    pthread_create(&threads[t], &attr, query_job, (void *) &obj->shards[t]);
  }

  /* Wait for all threads to finish */
  for(t = 0; t < obj->num_shards; t++) {
    pthread_join(threads[t], &status);
  }

  /* Find out which shard (if any) returned the data */
  for (t = 0; t < obj->num_shards; t++) {
    if (!obj->shards[t].empty) {
      result_shard = t;
      break;
    }
  }

  free(threads);
  return result_shard;
}

/** Disconnects from all shards with an open connection.
 *  @param[in,out] obj shardeddb_t struct with the object information
 */
static void disconnect_all_shards(shardeddb_t *obj) {
  int i;
  for (i = 0; i < obj->num_shards; i++) {
    if (obj->shards[i].connection != NULL) {
      mysql_close(obj->shards[i].connection);
    }
  }
}

/*******************************************************************************
 Ruby interface code
*******************************************************************************/

/** @privatesection */

/** Auxiliary method to calculate a difference between two times
 *  @param[in] start_time start time
 *  @param[in] end_time end time
 *  @return difference between both times, in milliseconds
 */
static double time_diff(struct timeval start_time, struct timeval end_time) {
  return ((end_time.tv_sec - start_time.tv_sec)  * 1000 +
     ((double)end_time.tv_usec - start_time.tv_usec) / 1000);
}

/** Auxiliary method to handle appending results to an existing result set,
 *  as well as perform the conversion between the MySQL API result sets and Ruby
 *  result sets.
 *  @param[in,out] existing an existing Ruby array, empty or otherwise
 *  @param[in] shard the shard containing the results to be appended
 */
void append_results(VALUE existing, shard_data_t* shard) {
  unsigned int num_fields;
  my_ulonglong num_rows;
  MYSQL_ROW row;
  MYSQL_FIELD *fields;
  VALUE hash, key, value, table, field_name;
  int i;

  /* Create Ruby structures for results */
  num_fields = mysql_num_fields(shard->results);
  num_rows = mysql_num_rows(shard->results);
  fields = mysql_fetch_fields(shard->results);

  while ((row = mysql_fetch_row(shard->results))) {
    hash = rb_hash_new();
    rb_ary_push(existing, hash);
    for (i = 0; i < num_fields; i++) {
      rb_hash_aset(hash, rb_str_new2(fields[i].name), row[i] != NULL ?
          rb_str_new2(row[i]) : Qnil);
    }
  }
  mysql_free_result(shard->results);
}

/** Destructor for the AdsDb class
 * @param[in] p the shardeddb_struct associated with the AdsDb instance to be
 * destroyed
 */
static void shardeddb_free(void *p) {
  shardeddb_t *shardeddb_struct = (shardeddb_t *)p;
  disconnect_all_shards(shardeddb_struct);
  free(shardeddb_struct->shards);
  free(shardeddb_struct);
  mysql_library_end();
}

/** @publicsection */

/** Initializer for the AdsDb class
 *  @param[in] self the AdsDb class object.
 *  @param[in] query_type multi-shard access semantics (one of the values in
 *             AdsDb::QueryType.
 *  @param[in] user the database username.
 *  @param[in] password the database password for this username.
 *  @param[in] host hostname "template" for the several shards. Must be a
 *             string with an optional placeholder for the shard number, e.g.
 *             "dbhost%d" for dbhost0, dbhost1, etc.
 *  @param[in] db database name "template" for the several shards. Must be
 *             a string with an optional placeholder for the shard number, e.g.
 *             "db%d" for db0, db1, etc.
 *  @param[in] num_shards the number of shards in the sharded database.
 *  @return the created AdsDb object
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def initialize(query_type, host, db, user, password, num_shards)
 *  @endcode
 */
static VALUE rb_shardeddb_new(VALUE self, VALUE query_type, VALUE user,
    VALUE password, VALUE host, VALUE db, VALUE num_shards) {
  shardeddb_t *shardeddb_struct;
  shard_data_t *shard;
  VALUE obj, debug_info;
  int i;

  obj = Data_Make_Struct(self, shardeddb_t, NULL, shardeddb_free,
      shardeddb_struct);

  if (NUM2DBL(num_shards) >= pow((double)10, (double)SHARD_DIGITS)) {
    rb_raise(sharded_db_error, ERROR_TOO_MANY_SHARDS);
  }

  shardeddb_struct->num_shards = NUM2INT(num_shards);
  shardeddb_struct->host = StringValuePtr(host);
  shardeddb_struct->db = StringValuePtr(db);
  shardeddb_struct->user = StringValuePtr(user);
  shardeddb_struct->password = StringValuePtr(password);
  shardeddb_struct->active_shard = -1;
  shardeddb_struct->shards = malloc(sizeof(shard_data_t) *
      shardeddb_struct->num_shards);
  shardeddb_struct->query_type = NUM2INT(query_type);

  for (i = 0; i < shardeddb_struct->num_shards; i++) {
    shard = &shardeddb_struct->shards[i];
    shard->db_info = shardeddb_struct;
    shard->id = i;
    shard->empty = 1;
    shard->connection = NULL;
    shard->results = NULL;
  }

  if (shardeddb_struct->query_type >= 0 ||
      shardeddb_struct->query_type <= INVALID_SHARD) {
      rb_raise(sharded_db_error, ERROR_QUERY_TYPE);
  }

  debug_info = rb_ary_new();
  rb_iv_set(obj, "@debug_info", debug_info);

  init_libmysql();

  return obj;
}

/** Getter for num_shards
 *  @param[in] self the AdsDb object being used
 *  @return the value for num_shards
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def num_shards @endcode
 */
static VALUE rb_shardeddb_num_shards(VALUE self) {
  shardeddb_t *shardeddb_struct;
  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);
  return INT2NUM(shardeddb_struct->num_shards);
}

/** Getter for user
 *  @param[in] self the AdsDb object being used
 *  @return the value for user
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def user @endcode
 */
static VALUE rb_shardeddb_user(VALUE self) {
  shardeddb_t *shardeddb_struct;
  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);
  return rb_str_new2(shardeddb_struct->user);
}

/** Getter for password
 *  @param[in] self the AdsDb object being used
 *  @return the value for password
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def password @endcode
 */
static VALUE rb_shardeddb_password(VALUE self) {
  shardeddb_t *shardeddb_struct;
  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);
  return rb_str_new2(shardeddb_struct->password);
}

/** Getter for debug_info
 *  @param[in] self the AdsDb object being used
 *  @return the value for debug_info
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def debug_info @endcode
 */
static VALUE rb_shardeddb_debug_info(VALUE self) {
  return rb_iv_get(self, "@debug_info");
}

/** Connects to the appropriate database shards.
 *  @param[in] self the AdsDb object being used
 *  @param[in] shard_num (optional, only valid for SINGLE_SHARD): shard on which
 *                       to perform the query
 *  @return Qnil (nil in Ruby), always
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def connect(shard_num = nil) @endcode
 */
static VALUE rb_shardeddb_connect(int argc, VALUE *argv, VALUE self) {
  shardeddb_t *shardeddb_struct;
  VALUE shard_num = Qnil;
  shard_data_t *shard;
  int shard_int;
  int i;

  /* Scan for one optional parameter */
  rb_scan_args(argc, argv, "01", &shard_num);

  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);

  switch (shardeddb_struct->query_type) {
    case ALL_SHARDS:
      connect_all_shards(shardeddb_struct);
      for (i = 0; i < shardeddb_struct->num_shards; i++) {
        shard = &shardeddb_struct->shards[i];
        if (shard->haveErrors) {
          rb_raise(sharded_db_error, shard->errorCode, shard->errorMessage);
        }
      }
      break;
    case UNKNOWN_SHARD:
      /* Do nothing, connections will be established as required, during the
         query phase. */
      break;
    case SINGLE_SHARD:
      shard_int = shard_num != Qnil ? NUM2INT(shard_num) : -1;
      if (shard_num != Qnil && shard_int >= 0 &&
          shard_int < shardeddb_struct->num_shards) {
        shard = &shardeddb_struct->shards[shard_int];
        connect_shard(shard);
        shardeddb_struct->active_shard = shard_int;
        if (shard->haveErrors) {
          rb_raise(sharded_db_error, shard->errorCode, shard->errorMessage);
        }
      } else {
        rb_raise(sharded_db_error, ERROR_SHARD_NUMBER);
      }
      break;
    default:
      rb_raise(sharded_db_error, ERROR_QUERY_TYPE);
  }

  return Qnil;
}

/** Executes an SQL query on the appropriate shards, returns results.
 *  Requires an open connection.
 *  @param[in] self the AdsDb object being used
 *  @param[in] query query to be executed
 *  @return array of hashes (@em TABLE.COLUMN as key, value as value) with the
 *          data
 *  @code [ { col1 => x1, col2 => y1, ...}, { col1 => x2, col2 => y2, ...}, ...]
 *  @endcode
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def query(query) @endcode
 */
static VALUE rb_shardeddb_query(VALUE self, VALUE query) {
  shardeddb_t *shardeddb_struct;
  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);
  char *query_string;
  int i, shard_num;
  shard_data_t *shard;
  VALUE results, debug_info, debug_entry;
  struct timeval start_time, end_time;

  gettimeofday(&start_time, NULL);
  query_string = StringValuePtr(query);

  switch (shardeddb_struct->query_type) {
    case ALL_SHARDS:
      query_all_shards(query_string, shardeddb_struct);

      for (i = 0; i < shardeddb_struct->num_shards; i++) {
        shard = &shardeddb_struct->shards[i];
        if (shard->haveErrors) {
          rb_raise(sharded_db_error, shard->errorCode, shard->errorMessage);
        }
      }
      break;
    case UNKNOWN_SHARD:
      /* Do we already know which shard the data is on? */
      shard_num = shardeddb_struct->active_shard;
      if (shard_num < 0) { /* No active shard */
        shard_num = connect_and_query_all_shards(query_string,
            shardeddb_struct);
        for (i = 0; i < shardeddb_struct->num_shards; i++) {
          shard = &shardeddb_struct->shards[i];
          if (shard->haveErrors) {
            rb_raise(sharded_db_error, shard->errorCode, shard->errorMessage);
          }
        }
      } else {
        query_shard(query_string, &shardeddb_struct->shards[shard_num]);
      }

      if (shard_num >= 0) {
        shardeddb_struct->active_shard = shard_num;
        shard = &(shardeddb_struct->shards[shard_num]);
      }
      break;
    case SINGLE_SHARD:
      if (shardeddb_struct->active_shard == -1) {
        rb_raise(sharded_db_error, ERROR_NOT_CONNECTED,
            "Database is not connected");
      }
      query_shard(query_string,
          &shardeddb_struct->shards[shardeddb_struct->active_shard]);
      shard = &shardeddb_struct->shards[shardeddb_struct->active_shard];
      if (shard->haveErrors) {
        rb_raise(sharded_db_error, shard->errorCode, shard->errorMessage);
      }
      break;
    default:
      rb_raise(sharded_db_error, ERROR_QUERY_TYPE);
  }

  results = rb_ary_new();
  for (i = 0; i < shardeddb_struct->num_shards; i++) {
    shard = &shardeddb_struct->shards[i];
    if (!shard->empty) {
      append_results(results, shard);
    }
  }

  debug_info = rb_iv_get(self, "@debug_info");
  debug_entry = rb_ary_new();
  rb_ary_push(debug_entry, query);
  gettimeofday(&end_time, NULL);
  rb_ary_push(debug_entry, rb_float_new(time_diff(start_time, end_time)));
  rb_ary_push(debug_info, debug_entry);
  return results;
}

/** Disconnect all open connections on the current AdsDb object.
 *  @param[in] self the AdsDb object being used
 *  @return Qnil (nil in Ruby), always
 *  @attention This C code generates the following method signature in Ruby:
 *  @code def disconnect @endcode
 */
static VALUE rb_shardeddb_disconnect(VALUE self) {
  shardeddb_t *shardeddb_struct;
  Data_Get_Struct(self, shardeddb_t, shardeddb_struct);
  disconnect_all_shards(shardeddb_struct);
  return Qnil;
}

/** Initializer for the Ruby extension. */
void Init_shardeddb() {
  /* Register ShardedDb class */
  VALUE shardeddb_class = rb_define_class("ShardedDb", rb_cObject);
  /* Register exception sub-class */
  sharded_db_error = rb_define_class_under(shardeddb_class, "Error",
      rb_eStandardError);
  /* Register query method sub-class */
  sharded_db_query_type = rb_define_class_under(shardeddb_class, "QueryType",
      rb_cObject);
  /* Register sharding method constants */
  rb_define_const(sharded_db_query_type, "SINGLE_SHARD", INT2NUM(SINGLE_SHARD));
  rb_define_const(sharded_db_query_type, "ALL_SHARDS", INT2NUM(ALL_SHARDS));
  rb_define_const(sharded_db_query_type, "UNKNOWN_SHARD",
      INT2NUM(UNKNOWN_SHARD));
  /* Register class methods */
  rb_define_singleton_method(shardeddb_class, "new", rb_shardeddb_new, 6);
  rb_define_method(shardeddb_class, "num_shards", rb_shardeddb_num_shards, 0);
  rb_define_method(shardeddb_class, "user", rb_shardeddb_user, 0);
  rb_define_method(shardeddb_class, "password", rb_shardeddb_password, 0);
  rb_define_method(shardeddb_class, "debug_info", rb_shardeddb_debug_info, 0);
  rb_define_method(shardeddb_class, "connect", rb_shardeddb_connect, -1);
  rb_define_method(shardeddb_class, "query", rb_shardeddb_query, 1);
  rb_define_method(shardeddb_class, "disconnect", rb_shardeddb_disconnect, 0);
}
